package com.bsj.power.pls.queue.service.impl.tactic;

import cn.hutool.core.exceptions.ExceptionUtil;
import com.bsj.power.common.util.DateUtils;
import com.bsj.power.common.util.SpringUtil;
import lombok.extern.slf4j.Slf4j;

import java.time.LocalDateTime;
import java.util.Arrays;
import java.util.LinkedList;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @author yinhao
 * @version 1.0
 * @description 融合消费数据线程
 * @date 2024/11/11
 */
@Slf4j
public class FusionDataConsume {

    private static List<byte[]> fusionList = new LinkedList<>();

    public boolean isRunning = true;
    private static Thread thread;
    private static RTKConsumeHandle rtkConsumeHandle = null;

    public void doStart() {
        rtkConsumeHandle = SpringUtil.getBean(RTKConsumeHandle.class);
        thread = new Thread(new Runnable() {
            @Override
            public void run() {
                while (isRunning) {
                    try {
                        LocalDateTime now = LocalDateTime.now();
                        //根据时间计算当前的包序号，按照大包编号分别为00，01，02，03，04，05
                        //按照每个小时分成6份，每十分钟对应一个编号
                        Integer pageNum = DateUtils.checkTimeByMinuteInt(now);
                        if (pageNum >= 6) {
                            //该时间范围时数据不发送,并且清除内存中的数据
                            if (fusionList.size() > 0 && !DateUtils.checkTimeByMinute(now)) {
                                fusionList.clear();
                            }
                            Thread.sleep(10);
                            continue;
                        }
                        if (fusionList.isEmpty()) {
                            //数据为空
                            Thread.sleep(10);
                            continue;
                        }
                        //将接收到的数据整合成一个数据
                        byte[] mergeByte = fusionList.stream().collect(Collectors.collectingAndThen(Collectors.toList(),
                                byteArrays -> mergeByteArrays(byteArrays)));
                        rtkConsumeHandle.handleCycle(mergeByte, pageNum);
                    } catch (Exception e) {
                        log.error(ExceptionUtil.getMessage(e));
                        continue;
                    }
                }
            }
        });
        log.info("初始化RTK数据融合线程...");
        thread.start();
    }

    public void doStop() {
        isRunning = false;
        if (thread != null) {
            thread.interrupt();
        }
    }

    /**
     * 依次写入数据
     *
     * @param bytes
     */
    public static void putData(byte[] bytes) {
        try {
            fusionList.add(bytes);
        } catch (Exception e) {
            log.error(ExceptionUtil.getMessage(e));
            return;
        }
    }

    /**
     * 计算了所有字节数组的总长度，然后创建了一个新的字节数组来存储合并后的数据
     * @param byteArrays
     * @return
     */
    private static byte[] mergeByteArrays(List<byte[]> byteArrays) {
        int totalLength = byteArrays.stream().mapToInt(arr -> arr.length).sum();
        byte[] result = new byte[totalLength];
        int currentIndex = 0;
        for (byte[] array : byteArrays) {
            System.arraycopy(array, 0, result, currentIndex, array.length);
            currentIndex += array.length;
        }
        return result;
    }
}
